package com.hub.realtime.common.utils;

import com.streamxhub.streamx.common.util.CommandUtils;

import java.io.File;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class FlinkUtils {

    public static String flinkVersion(String flinkHome)
    {
        Pattern flinkVersionPattern = Pattern.compile("^Version: (.*), Commit ID: (.*)$");
        AtomicReference<String> flinkVersion = new AtomicReference<>("");
        String libPath = flinkHome.concat("/lib");
        File[] distJar = new File(libPath).listFiles(x -> x.getName().matches("flink-dist_.*\\.jar"));
        if (distJar == null || distJar.length == 0) {
            throw new IllegalArgumentException("flink-dist jar 不存在 " + libPath+" 路径中");
        }
        if (distJar.length > 1) {
            throw new IllegalArgumentException("不能存在多个 flink-dist jar 在 " + libPath+" 路径中");
        }
        List<String> cmd = Arrays.asList(
                "cd ".concat(flinkHome),
                String.format(
                        "java -classpath %s org.apache.flink.client.cli.CliFrontend --version",
                        distJar[0].getAbsolutePath()
                )
        );
        CommandUtils.execute(cmd, versionInfo -> {
            Matcher matcher = flinkVersionPattern.matcher(versionInfo);
            if (matcher.find()) {
                flinkVersion.set(matcher.group(1));
            }
        });
        return flinkVersion.get();
    }

}
